﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.

using System.Globalization;
using System.Threading;
using System.Threading.Tasks;
using System.Fabric;
using System.Threading.Actors.Runtime;
using System.Text;
using System.Linq;
using System.Threading.Actors.Utilities;
using System.Threading.Actors.Helpers;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Threading.Actors.Languages;
using System.Diagnostics.Contracts;

namespace System.Threading.Actors.StatelessActor
{
    // Provides support for a "stateless" Windows Fabric actor.
    // The actor can still hold state, but it uses something besides WinFab replication to ensure the durability of that state.
    public class StatelessActorService : IStatelessServiceInstance, IActor
    {
        private ActorRawTcpListener m_listener;
        private StatelessServiceInitializationParameters initializationParameters;
        private const string EndpointName = "StatelessActorEndpoint";
        private string traceIdentifier = "StatelessActorService";
        private Uri m_listenUri;

        // ActorServiceHelpers for each sub-actor as distinguished by partition key/name
        private readonly ConcurrentDictionary<string, ActorServiceHelper> m_partitionSpecificHelpers = new ConcurrentDictionary<string, ActorServiceHelper>();
        
        // ActorServiceHelper for default sub-actor, i.e. the one with null/empty partition key/name
        private ActorServiceHelper m_defaultHelper;

        private Action<IActorState> m_methodInitializer; // Determines initial personality of actors
        private NonReplicatingStateCoordinator m_stateCoordinator; // Holder/coordinator for sub-actor states.
        private string m_serviceName;

        public StatelessActorService()
            : base()
        {
        }

        // Retrieve the ActorServiceHelper that corresponds to a partitionName, creating it if necessary.
        private ActorServiceHelper GetServiceHelper(string partitionName)
        {
            Contract.Ensures(Contract.Result<ActorServiceHelper>() != null);
            bool created = false;

            if (String.IsNullOrEmpty(partitionName))
            {
                if (m_defaultHelper == null)
                {
                    var state = m_stateCoordinator.GetActorState(null);
                    m_defaultHelper = new ActorServiceHelper(state, m_serviceName, m_methodInitializer);
                    created = true;
                }
                if (created)
                    ActorETWEventSource.Log.ActorCreated(m_serviceName, null);

                return m_defaultHelper;
            }
            
            ActorServiceHelper result;
            if (!m_partitionSpecificHelpers.TryGetValue(partitionName, out result))
            {
                lock (m_partitionSpecificHelpers)
                {
                    if (!m_partitionSpecificHelpers.TryGetValue(partitionName, out result))
                    {
                        var state = m_stateCoordinator.GetActorState(partitionName);
                        result = new ActorServiceHelper(state, m_serviceName, m_methodInitializer);
                        m_partitionSpecificHelpers[partitionName] = result;
                        created = true;
                    }
                }
                if (created)
                    ActorETWEventSource.Log.ActorCreated(m_serviceName, null);
            }
            else
			    Contract.Assert(result != null, "Partition names must be associated with an ActorServiceHelper, not null.");

            return result;              
        }

        /// <summary>
        /// This is the very first method called by the framework
        /// Provides all the initialization information
        /// No complex processing should be done here
        /// All heavy lifting should be done in Open
        /// </summary>
        /// <param name="initializationParameters"></param>
        public void Initialize(StatelessServiceInitializationParameters initializationParameters)
        {
            Utility.Assert(initializationParameters != null, "Windows Fabric guarantees that initializationParameters is never null");

            this.initializationParameters = initializationParameters;
            Console.WriteLine(string.Format(CultureInfo.InvariantCulture, "{0}, Initialize, ServiceName: {1}, InstanceID: {2}, PartitionID: {3}", this.traceIdentifier, initializationParameters.ServiceName, initializationParameters.InstanceId, initializationParameters.PartitionId));

            // Convert the initialization string into a temporary key-value store (i.e., dictionary)
            var initializationString = Encoding.UTF8.GetString(initializationParameters.InitializationData);
            var parts = initializationString.Split(';');
            var initializationDictionary = new Dictionary<string, string>();
            foreach (var part in parts)
            {
                var keyAndValue = part.Split('=');
                if (keyAndValue.Length != 2) continue; // mal-formed
                initializationDictionary[keyAndValue[0]] = keyAndValue[1];
            }

            // Grab the service name
            m_serviceName = initializationParameters.ServiceName.ToString();

            // Figure out our state type
            // (Right now, we only support InMemory, but others will be added.)
            StatelessActorStateTypes storageType = StatelessActorStateTypes.InMemory; // by default
            string storageString;
            if (initializationDictionary.TryGetValue("StateType", out storageString))
            {
                storageType = (StatelessActorStateTypes) Enum.Parse(typeof(StatelessActorStateTypes), storageString, ignoreCase: true);
            }

            // .. and initialize our state coordinator
            m_stateCoordinator = new NonReplicatingStateCoordinator(m_serviceName, storageType);

            // Grab our personality (defaults to Empty) and create a method initializer
            string personality;
            if (!initializationDictionary.TryGetValue("Personality", out personality))
            {
                personality = "Empty";
            } 
            if (personality.Equals("Empty", StringComparison.InvariantCultureIgnoreCase))
            {
                m_methodInitializer = null;
            }
            else if (personality.Equals("List", StringComparison.InvariantCultureIgnoreCase))
            {
                m_methodInitializer = state => ListHelperMethods.LoadActorMethods(state);
            }
            else if (personality.Equals("Dictionary", StringComparison.InvariantCultureIgnoreCase))
            {
                m_methodInitializer = state => DictionaryActorMethods.LoadActorMethods(state);
            }
            else throw new InvalidOperationException("Unknown personality type " + personality);

            // Now initialize our default service helper
            var defaultState = m_stateCoordinator.GetActorState(null);
            m_defaultHelper = new ActorServiceHelper(defaultState, m_serviceName, m_methodInitializer);
        }

        /// <summary>
        /// OpenAsync is called when the replica is going to be actually used
        /// The service is expected to return a Task that prepares the service for execution and at the end, returns the endpoint Uri as the Task.Result.
        /// This is used to inform Windows Fabric about the endpoint the service is listening to, so that Windows Fabric can advertise it accordingly.
        /// </summary>
        public Task<string> OpenAsync(IStatelessServicePartition partition, CancellationToken cancellationToken)
        {
            // TODO: tcs is unnecessary.  Just have the task below return a string, and return that task to the caller.
            // In fact, doing this async at all is probably unnecessary -- just do it synchronously and return Task.FromResult(listenUri.ToString()).
            var tcs = new TaskCompletionSource<string>();
            Task.Factory.StartNew(() =>
            {
                try
                {
                    var serviceManifestEndpoint = this.initializationParameters.CodePackageActivationContext.GetEndpoints()[EndpointName];
                    m_listenUri = new UriBuilder(Utility.GetScheme(serviceManifestEndpoint.Protocol), NetworkConfiguration.GetIPAddressOrHostName(false, NetworkConfiguration.PreferIPv6), serviceManifestEndpoint.Port, this.initializationParameters.PartitionId.ToString() + "_" + this.initializationParameters.InstanceId.ToString(CultureInfo.InvariantCulture)).Uri;
                    // Created Uri has the partitionID and instanceID making it uniquely identifiable.
                    // beginOpen is called asynchronously
                    // If the open is successful, the endpoint uri is returned to Windows Fabric as the Task.Result
                    m_listener = ActorRawTcpListener.GetListener(m_listenUri);
                    m_listener.RegisterService(m_listenUri, this);
                    Console.WriteLine(string.Format(CultureInfo.InvariantCulture, "{0}, OpenAsync, Listening to {1}", this.traceIdentifier, m_listenUri));
                    Console.WriteLine();
                    tcs.SetResult(m_listenUri.ToString());
                }
                catch (Exception e)
                {
                    tcs.SetException(e);
                }
            }
            );
            return tcs.Task;
        }

        /// <summary>
        /// CloseAsync is called when the replica is going to be closed
        /// </summary>
        public Task CloseAsync(CancellationToken cancellationToken)
        {
            return Task.Factory.StartNew(() =>
            {
                Console.WriteLine(string.Format(CultureInfo.InvariantCulture, "{0}, CloseAsync, Closing {1}", this.traceIdentifier, this.initializationParameters.ServiceName));
                if (m_stateCoordinator != null) m_stateCoordinator.Dispose();
                if(m_listener != null && m_listenUri != null)
                    m_listener.UnregisterService(m_listenUri);
            });
        }

        /// <summary>
        /// Abort is called when the replica is being forced closed
        /// </summary>
        public void Abort()
        {
            Console.WriteLine(string.Format(CultureInfo.InvariantCulture, "{0}, Abort, Abort {1}", this.traceIdentifier, this.initializationParameters.ServiceName));
            if (m_stateCoordinator != null) m_stateCoordinator.Dispose();
            if (m_listener != null && m_listenUri != null)
                m_listener.UnregisterService(m_listenUri);
        }

        #region IActor interface implementation

        // Returns an ActorServiceResult encoded as a JSON string, ASR.Result is object
        public Task<string> CallMethod(string clientId, int clientSequenceNumber, string partitionName, string methodName, string[] parameters)
        {
            return GetServiceHelper(partitionName).CallMethod(clientId, clientSequenceNumber, methodName, parameters);
        }

        // Returns an ActorServiceResult encoded as a JSON string, ASR.Result is bool
        public Task<string> AddAssembly(int clientSequenceNumber, string partitionName, string assemblyName, byte[] assemblyBytes)
        {
            return GetServiceHelper(partitionName).AddAssembly(clientSequenceNumber, assemblyName, assemblyBytes);
        }

        // Returns an ActorServiceResult encoded as a JSON string, ASR.Result is null
        public string Subscribe(int clientSequenceNumber, string partitionName, string eventType, IPublicationCallbackContract context)
        {
            return GetServiceHelper(partitionName).Subscribe(clientSequenceNumber, eventType, context);
        }

        // Returns an ActorServiceResult encoded as a JSON string, ASR.Result is null
        public string Unsubscribe(int clientSequenceNumber, string partitionName, string eventType, IPublicationCallbackContract context)
        {
            return GetServiceHelper(partitionName).Unsubscribe(clientSequenceNumber, eventType, context);
        }

        // Returns an ActorServiceResult encoded as a JSON string, ASR.Result is null
        public string UnsubscribeAll(int clientSequenceNumber, string partitionName, IPublicationCallbackContract context)
        {
            return GetServiceHelper(partitionName).UnsubscribeAll(clientSequenceNumber, context);
        }

        // Returns an ActorServiceResult encoded as a JSON string, ASR.Result is the service's address
        // (e.g., "fabric:/actor/list/mylist") in string form.
        public string Ping(int clientSequenceNumber, string partitionName)
        {
            var myAddress = initializationParameters.ServiceName.ToString();
            return SerializationHelper.Serialize(ActorServiceResult.FromResult(clientSequenceNumber, myAddress));
        }

        public Task<string> AddMethod(int clientSequenceNumber, string partitionName, int language, string methodName, string methodImpl)
        {
            return GetServiceHelper(partitionName).AddMethod(clientSequenceNumber, (SupportedLanguages)language, methodName, methodImpl);
        }

        public string Configure(int clientSequenceNumber, string partitionName, string configurationString)
        {
            return GetServiceHelper(partitionName).Configure(clientSequenceNumber, configurationString);
        }

        public string DeleteSelf(int clientSequenceNumber, string partitionName)
        {
            String ret = GetServiceHelper(partitionName).DeleteSelf(clientSequenceNumber);
            if (partitionName != null)
            {
                lock (m_partitionSpecificHelpers)
                {
                    ActorServiceHelper serviceHelper = null;
                    if (!m_partitionSpecificHelpers.TryRemove(partitionName, out serviceHelper))
                    {
                        // We lost a race to remove this.  Let's assume this is a safe race to lose.
                        //throw new InvalidOperationException(String.Format("Couldn't delete stateless partitioned actor state for actor {0} partition {1}.  Did you already delete it?",
                        //    m_serviceName, partitionName));
                    }
                }
            }
            return ret;
        }

        #endregion

    }
}
